package net.bwie.realtime.vehicle.a;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @Author: FuHe
 * @Date: 2025/5/28
 */
public class VehicleBatteryHealthJob {
    public static void main(String[] args) {
        // 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 输入表  input 映射到kafka消息队列
        createInputTable(tabEnv);
        // 数据处理
        Table reportTable = handle(tabEnv);
        reportTable.execute().print();
        // 输出表  output  映射到clickhouse表
//        createOutputTable(tabEnv);
        // 保存数据
//        saveToClickhouse(tabEnv, reportTable);
    }

    private static Table handle(TableEnvironment tabEnv) {
        Table table = tabEnv.sqlQuery("SELECT\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    AVG(soh) AS avg_soh,\n" +
                "    COUNT(*) AS vehicle_count\n" +
                "FROM TABLE(\n" +
                "             TUMBLE(TABLE battery_health, DESCRIPTOR(`timestamp`), INTERVAL '5' MINUTES))\n" +
                "                 GROUP BY window_start, window_end");
        Table table1 = tabEnv.sqlQuery("SELECT\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    SUM(CASE WHEN soh >= 90 THEN 1 ELSE 0 END) AS excellent,\n" +
                "    SUM(CASE WHEN soh >= 80 AND soh < 90 THEN 1 ELSE 0 END) AS good,\n" +
                "    SUM(CASE WHEN soh >= 70 AND soh < 80 THEN 1 ELSE 0 END) AS fair,\n" +
                "    SUM(CASE WHEN soh < 70 THEN 1 ELSE 0 END) AS poor\n" +
                "FROM TABLE(\n" +
                "             TUMBLE(TABLE battery_health, DESCRIPTOR(`timestamp`), INTERVAL '15' MINUTES))\n" +
                "                 GROUP BY window_start, window_end");
        return table;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE battery_health (\n" +
                "                                vin STRING,\n" +
                "                                `timestamp` TIMESTAMP(3),\n" +
                "                                soh DOUBLE,\n" +
                "                                temp DOUBLE,\n" +
                "                                cycles INT,\n" +
                "                                WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'vehicle-battery-data',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                "    'format' = 'json'\n" +
                ")");
    }

    private static TableEnvironment getTableEnv() {
        // 环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        // 时区设置
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        // 并发设置
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态保存时间
        configuration.setString("table.exec.state.ttl", "5s");
        // 返回值
        return tabEnv;
    }
}
